package cn.com.bluemoon.daps.sync.service.impl;

import cn.com.bluemoon.daps.common.domain.UserInfoHolder;
import cn.com.bluemoon.daps.common.enums.SyncStatus;
import cn.com.bluemoon.daps.common.enums.SyncToEnv;
import cn.com.bluemoon.daps.common.exception.DapException;
import cn.com.bluemoon.daps.common.mp.IBmService;
import cn.com.bluemoon.daps.common.toolkit.BmAssetUtils;
import cn.com.bluemoon.daps.common.toolkit.SshUtils;
import cn.com.bluemoon.daps.model.entity.DapDataModelTable;
import cn.com.bluemoon.daps.sync.base.CompareResult;
import cn.com.bluemoon.daps.sync.base.ModifyType;
import cn.com.bluemoon.daps.sync.base.SqlCompareResult;
import cn.com.bluemoon.daps.sync.config.PtServerConfig;
import cn.com.bluemoon.daps.sync.core.sync.GlobalConfig;
import cn.com.bluemoon.daps.sync.core.sync.TableModel;
import cn.com.bluemoon.daps.sync.core.sync.compare.ModelCompareClient;
import cn.com.bluemoon.daps.sync.core.sync.executor.SqlExecutor;
import cn.com.bluemoon.daps.sync.core.sync.executor.SqlExecutors;
import cn.com.bluemoon.daps.sync.core.sync.log.EvenType;
import cn.com.bluemoon.daps.sync.core.sync.log.SyncEnvLog;
import cn.com.bluemoon.daps.sync.core.sync.log.SyncEnvLogEvent;
import cn.com.bluemoon.daps.sync.entity.DapSyncDevEnvTask;
import cn.com.bluemoon.daps.sync.feignclient.DapModelModuleFacade;
import cn.com.bluemoon.daps.sync.mapper.DapSyncDevEnvTaskMapper;
import cn.com.bluemoon.daps.sync.service.DapModel2DevSyncService;
import cn.com.bluemoon.daps.sync.service.DapSyncBasicService;
import cn.com.bluemoon.daps.sync.service.DapSyncDevEnvTaskService;
import cn.com.bluemoon.daps.sync.vo.model2dev.SyncTableInfoOut;
import cn.com.bluemoon.daps.sync.vo.model2dev.SyncTablePageVo;
import cn.com.bluemoon.daps.sync.vo.model2dev.SyncTableTaskOut;
import cn.com.bluemoon.daps.sync.vo.model2dev.SyncTableTaskPageVo;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.event.EventListener;
import org.springframework.stereotype.Service;
import reactor.util.function.Tuple2;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;

import static cn.com.bluemoon.daps.common.enums.SyncStatus.SYNC_ALL;

/**
 * 开发同步（模型同步到开发库/测试库）
 * 1.获取元数据
 * 2.获取模型数据
 * 3.进行同步对比工具，生成sql/ptosc
 * 4.进入执行器，完成执行，修改同步任务信息
 * 5.发送邮件，记录日志
 *
 * @author Jarod.Kong
 * @date 2021/6/16 21:17
 */
@Slf4j
@Service
public class DapModel2DevSyncServiceImpl implements DapModel2DevSyncService {

    @Autowired
    private DapSyncBasicService syncBasicService;

    @Autowired
    private DapModelModuleFacade modelModuleFacade;

    @Autowired
    private DapSyncDevEnvTaskService syncDevEnvTaskService;

    @Resource
    private PtServerConfig ptServerConfig;

    /**
     * 模型表数据同步到开发库、测试库前的对比结果SQL
     *
     * @param modelId   模型id
     * @param syncToEnv 同步到的环境
     * @param tableIds  所选的同步表
     * @return 对比结果
     */
    @Override
    public List<CompareResult<SqlCompareResult>> syncPreCompareResult(String modelId, SyncToEnv syncToEnv, List<String> tableIds) {
        BmAssetUtils.isTrue(SyncToEnv.DEV.equals(syncToEnv) || SyncToEnv.TEST.equals(syncToEnv), "目前允许同步开发库和同步测试库");
        // 判断表id是否处于任务中
        this.syncPreCheckTableStatus(modelId, tableIds);
        // 表id转表英文名
        List<String> tableNames = findModelTableNameByModelIdAndTableIds(modelId, tableIds);
        GlobalConfig globalConfig = syncBasicService.buildSyncGlobalConfig(modelId, syncToEnv, null, true);
        globalConfig.setTableIdentity(new GlobalConfig.TableIdentity(tableIds, tableNames));
        List<Map<ModifyType.ModifyLevel, List<CompareResult<SqlCompareResult>>>> compareR = syncCompare0(globalConfig);
        List<CompareResult<SqlCompareResult>> res = compareR.stream().flatMap(m -> m.entrySet().stream()).flatMap(e -> e.getValue().stream()).collect(Collectors.toList());
        return res;
    }

    /**
     * 判断表id是否处于任务中
     *
     * @param modelId  模型
     * @param tableIds 所选表id
     */
    private void syncPreCheckTableStatus(String modelId, List<String> tableIds) {
        if (tableIds == null || tableIds.isEmpty()) {
            return;
        }
        LambdaQueryWrapper<DapSyncDevEnvTask> eq = syncDevEnvTaskService.bmLambdaQuery().eq(DapSyncDevEnvTask::getModelId, modelId).in(DapSyncDevEnvTask::getTableId, tableIds)
                .eq(DapSyncDevEnvTask::getSyncStatus, SyncStatus.SYNC_DOING);
        List<DapSyncDevEnvTask> syncDoingTasks = syncDevEnvTaskService.list(eq);
        if (!syncDoingTasks.isEmpty()) {
            List<String> syncDoingTableIds = syncDoingTasks.stream().map(DapSyncDevEnvTask::getTableId).distinct().collect(Collectors.toList());
            List<String> tableNames = findModelTableNameByModelIdAndTableIds(modelId, syncDoingTableIds);
            throw new DapException("所选表处于同步中，如下：\n" + String.join("\n", tableNames));
        }
    }

    /**
     * 表id获取表名列表
     *
     * @param modelId  模型
     * @param tableIds 表ids
     * @return 表名列表
     */
    private List<String> findModelTableNameByModelIdAndTableIds(String modelId, List<String> tableIds) {
        // 表id转为表名s
        List<DapDataModelTable> tables = modelModuleFacade.queryModelTablesByTableIds(modelId, true, tableIds);
        List<String> tableNames = tables.stream().map(DapDataModelTable::getEngName).collect(Collectors.toList());
        return tableNames;
    }

    /**
     * 模型表数据同步到开发库、测试库
     * 对比结果
     * 异步执行
     *
     * @param modelId   模型id
     * @param syncToEnv 同步到的环境
     * @param tableIds  所选的同步表
     * @param userInfo  操作人
     * @return 同步结果 都是true
     */
    @Override
    public boolean async(String modelId, SyncToEnv syncToEnv, List<String> tableIds, UserInfoHolder.UserInfo userInfo) {
        return sync(modelId, syncToEnv, tableIds, userInfo, true);
    }

    /**
     * 模型表数据同步到开发库、测试库
     * 对比结果
     *
     * @param modelId   模型id
     * @param syncToEnv 同步到的环境
     * @param tableIds  所选的同步表
     * @param userInfo  操作人
     * @param async     是否异步执行 true开启异步执行，返回结果都是true，false会根据实际执行情况返回结果
     * @return 同步结果
     */
    @Override
    public boolean sync(String modelId, SyncToEnv syncToEnv, List<String> tableIds, UserInfoHolder.UserInfo userInfo, boolean async) {
        BmAssetUtils.isTrue(SyncToEnv.DEV.equals(syncToEnv) || SyncToEnv.TEST.equals(syncToEnv), "目前允许同步开发库和同步测试库");
        // 判断表id是否处于任务中
        this.syncPreCheckTableStatus(modelId, tableIds);
        // 表id转表英文名
        List<String> tableNames = findModelTableNameByModelIdAndTableIds(modelId, tableIds);
        GlobalConfig globalConfig = syncBasicService.buildSyncGlobalConfig(modelId, syncToEnv, userInfo, true);
        globalConfig.setTableIdentity(new GlobalConfig.TableIdentity(tableIds, tableNames));
        if (async) {
            CompletableFuture.runAsync(() -> doSync0(globalConfig));
            return true;
        } else {
            return doSync0(globalConfig);
        }
    }

    /**
     * 获取开发同步列表分页数据
     *
     * @param searchVo 查询条件
     * @return list
     */
    @Override
    public IPage<SyncTableInfoOut> querySyncList(SyncTablePageVo searchVo) {
        modelModuleFacade.getByIdAsset(searchVo.getModelId(), "模型不存在！！");
        // 2021/6/17 结合模型最新表+同步任务记录 + 根据表修改时间排序desc
        List<DapDataModelTable> modelTables = modelModuleFacade.queryModelTables(searchVo.getModelId(), true);
        List<SyncTableInfoOut> syncTableInfos = modelTables.stream()
                .map(t -> {
                    SyncTableInfoOut out = SyncTableInfoOut.builder()
                            .modelId(t.getDataModelId()).tableId(t.getId()).tableName(t.getEngName()).tableNameZh(t.getName())
                            .updateBy(t.getUpdateBy()).updateTime(t.getUpdateTime())// 2021/7/8 v1.3.1存在疑问，表的修改时间无法确定表相关信息是否修改了
                            .build();
                    DapSyncDevEnvTaskMapper mapper = syncDevEnvTaskService.getBaseMapper(DapSyncDevEnvTaskMapper.class);
                    List<DapSyncDevEnvTask> tasks = mapper.queryDevSyncUnionAllTestSyncNewestOne(t.getDataModelId(), t.getId());
                    // 拿最新一条的执行人
                    if (!tasks.isEmpty()) {
                        tasks.stream().min((o1, o2) -> o2.getCreateTime().compareTo(o1.getCreateTime()))
                                .ifPresent(d -> out.setNewestSyncBy(d.getSyncExecBy()));
                    }
                    for (DapSyncDevEnvTask d : tasks) {
                        if (SyncToEnv.DEV.equals(d.getSyncEnv())) {
                            out.setDevSyncStatus(d.getSyncStatus().getCode());
                            out.setDevSyncTime(d._getSyncTime(false));
                        }
                        if (SyncToEnv.TEST.equals(d.getSyncEnv())) {
                            out.setTestSyncStatus(d.getSyncStatus().getCode());
                            out.setTestSyncTime(d._getSyncTime(false));
                        }
                    }
                    return out;
                }).filter(t -> {
                    String keywordTrim = searchVo.getKeywordTrim();
                    boolean matchName = true;
                    if (StringUtils.isNotBlank(keywordTrim)) {
                        boolean matchNameEn = t.getTableName().toLowerCase().contains(keywordTrim);
                        boolean matchNameZh = t.getTableNameZh().toLowerCase().contains(keywordTrim);
                        matchName = (matchNameEn || matchNameZh);
                    }
                    boolean matchDevStatus = true;
                    SyncStatus devStatus = searchVo.getDevStatus();
                    if (devStatus != null && !SYNC_ALL.equals(devStatus)) {
                        matchDevStatus = devStatus.equals(SyncStatus.valOf(t.getDevSyncStatus()));
                    }
                    boolean matchTestStatus = true;
                    SyncStatus testStatus = searchVo.getTestStatus();
                    if (testStatus != null && !SYNC_ALL.equals(testStatus)) {
                        matchTestStatus = testStatus.equals(SyncStatus.valOf(t.getTestSyncStatus()));
                    }
                    return matchName & matchDevStatus & matchTestStatus;
                }).collect(Collectors.toList());
        IPage<SyncTableInfoOut> page = IBmService.logicPage(syncTableInfos, searchVo.getCurrent(), searchVo.getPageSize());
        return page;
    }

    /**
     * 查看表同步日志
     *
     * @param searchVo#getTableId 表id 对应开发同步行id
     *                            1@param searchVo#getType    查看表同步类型，1 查看开发库日志 2查看测试库日志, 默认为1
     * @return page
     */
    @Override
    public IPage<SyncTableTaskOut> viewSyncTasks(SyncTableTaskPageVo searchVo) {
        modelModuleFacade.getByIdAsset(searchVo.getModelId(), "模型不存在！！");
        @SuppressWarnings("unchecked") LambdaQueryWrapper<DapSyncDevEnvTask> eq = syncDevEnvTaskService.bmLambdaQuery()
                .eq(DapSyncDevEnvTask::getTableId, searchVo.getTableId())
                .eq(DapSyncDevEnvTask::getModelId, searchVo.getModelId())
                .eq(DapSyncDevEnvTask::getSyncEnv, searchVo.getType())
                .orderByDesc(DapSyncDevEnvTask::getCreateTime, DapSyncDevEnvTask::getUpdateTime);
        @SuppressWarnings("unchecked") IPage<DapSyncDevEnvTask> page = syncDevEnvTaskService.page(searchVo.getMpPage(), eq);
        return page.convert(t -> {
            SyncTableTaskOut out = new SyncTableTaskOut();
            out.setDdl(t.getSyncDdl());
            out.setRowId(t.getId());
            out.setModelId(t.getModelId());
            out.setTableId(t.getTableId());
            out.setSyncBy(t.getSyncExecBy());
            out.setSyncEnv(t.getSyncEnv().getCode());
            out.setSyncStatus(t.getSyncStatus().getCode());
            out.setSyncTime(t._getSyncTime(false));
            return out;
        });
    }

    /**
     * 根据表任务id 表${dap_sync_dev_env_task.id}
     * 1.当前任务id状态
     * 1）为同步中，根据sync_log_path进行ssh拉取实时数据
     * 2）为已同步、同步失败，直接拿sync_log返回
     *
     * @param syncTaskId id
     * @return log
     */
    @Override
    public String viewSyncTaskLog(String syncTaskId) {
        LambdaQueryWrapper<DapSyncDevEnvTask> eq = syncDevEnvTaskService.bmLambdaQuery()
                .eq(DapSyncDevEnvTask::getId, syncTaskId)
                .last("limit 1");
        DapSyncDevEnvTask envTask = syncDevEnvTaskService.getOne(eq);
        BmAssetUtils.notNull(envTask, "获取表同步日志失败");
        BmAssetUtils.isTrue(envTask.getSyncStatus() != null, "同步状态不可为空");
        List<String> catLogContent = new ArrayList<>(1);
        if (SyncStatus.SYNC_DOING.equals(envTask.getSyncStatus())) {
            // ssh log
            String catLog = String.format("cat %s", envTask.getSyncLogPath());
            String logData = SshUtils.client().sshInfo(ptServerConfig.getPtSsh()).runScriptAsset(catLog);
            catLogContent.add(logData);
        } else {
            catLogContent.add(envTask.getSyncLog());
            log.info("直接获取DB中日志数据");
        }
        return String.join(System.lineSeparator(), catLogContent);
    }

    /**
     * 所选的模型表对比开发、测试库表
     *
     * @param globalConfig 全局配置
     * @return list
     */
    private List<Map<ModifyType.ModifyLevel, List<CompareResult<SqlCompareResult>>>> syncCompare0(GlobalConfig globalConfig) {
        Tuple2<List<TableModel>, List<TableModel>> localAndMetadataTableModel = syncBasicService.getLocalAndMetadataTableModel(globalConfig);
        List<TableModel> modelV1 = localAndMetadataTableModel.getT1();
        List<TableModel> modelV2 = localAndMetadataTableModel.getT2();
        globalConfig.setCheckSqlLayer(true);
        globalConfig.setCanDropDbTable(false);
        List<Map<ModifyType.ModifyLevel, List<CompareResult<SqlCompareResult>>>> compareR = ModelCompareClient
                .doCompare(ModelCompareClient.BizType.SQL, globalConfig, modelV2, modelV1);
        log.info("模型导入开发库（开发同步）：{}, size={}", compareR, compareR.size());
        return compareR;
    }

    /**
     * 异步执行同步模型表到开发库、测试库
     * 所选表进行同步
     *
     * @param globalConfig 全局配置
     * @return 同步结果
     */
    private boolean doSync0(GlobalConfig globalConfig) {
        List<Map<ModifyType.ModifyLevel, List<CompareResult<SqlCompareResult>>>> res = syncCompare0(globalConfig);
        BmAssetUtils.isFalse(res.isEmpty(), "对比结果为空，不进行同步");
        // 根据模型对应的数据源中的开关（是否开启ptosc同步）
        Boolean isPtosc = globalConfig.getDataSource().getIsPtosc();
        SqlExecutor.ExecutorType type;
        if (isPtosc == null || Boolean.TRUE.equals(isPtosc)) {
            type = SqlExecutor.ExecutorType.PT;
        } else {
            type = SqlExecutor.ExecutorType.JDBC;
        }
        // 异步提交等待返回
        return SqlExecutors.use(type, globalConfig).async().sqlResult(res).execute();
    }

    @EventListener(classes = SyncEnvLogEvent.class)
    public void syncEnvLogEventHandler(SyncEnvLogEvent envLogEvent) {
        SyncEnvLog syncEnvLog = envLogEvent.getSource();
        log.info("处理消息，TYPE={}，模型：{}，表id：{}", envLogEvent.getType(), syncEnvLog.getModelId(), syncEnvLog.getTableId());
        if (EvenType.SAVE_OR_UPDATE.equals(envLogEvent.getType())) {
            DapSyncDevEnvTask task = new DapSyncDevEnvTask();
            BeanUtils.copyProperties(syncEnvLog, task);
            task.setId(syncEnvLog.getKey());
            task.setSyncEnv(syncEnvLog.getSyncEnv());
            task.setSyncStatus(syncEnvLog.getSyncStatus());
            task.setSyncDdl(String.join(";\n", syncEnvLog.getSyncDdl()));
            task.setSyncPtsql(String.join(",", syncEnvLog.getSyncPtsql()));
            task.setSyncExecBy(syncEnvLog.getSyncExecBy());
            task.setSyncLog(String.join(System.lineSeparator(), syncEnvLog.getSyncLog()));
            task.setSyncPreTableCreateSql(syncEnvLog.getTableCreateSql());
            boolean status = syncDevEnvTaskService.saveOrUpdate(task);
            log.info("新增/更新任务执行情况：status={}", status);
        } else {
            // APPEND_UPDATE
            DapSyncDevEnvTask task = syncDevEnvTaskService.getByIdAsset(syncEnvLog.getKey(), "获取开发同步任务表任务信息失败");
            List<String> preSyncLog = syncEnvLog.getSyncLog();
            if (preSyncLog != null && !preSyncLog.isEmpty()) {
                task.setSyncLog(task.getSyncLog() + System.lineSeparator() + String.join(System.lineSeparator(), preSyncLog));
            }
            task.setSyncStatus(syncEnvLog.getSyncStatus());
            task.setSyncEndTime(syncEnvLog.getSyncEndTime());
            task.setSyncScript(JSON.toJSONString(syncEnvLog.getSyncScript()));
            boolean status = syncDevEnvTaskService.updateById(task);
            log.info("更新任务执行情况：status={}", status);
            // 2021/11/2 清除表字段名称
            modelModuleFacade.delTableRenames(task.getTableId(), null);
            // send mail
            log.info("开始发送结果邮件消息");
            syncBasicService.sendDevEmail(syncEnvLog);
        }
    }

}
